package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class test8 {
    public static void main(String[] args) throws Exception {
        //流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //检查点
        env.enableCheckpointing(5000);
        //kafka读取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test8");
        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));

        SingleOutputStreamOperator<String> orderInfo = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));

        //orderInfo.print();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SingleOutputStreamOperator<Order_info> orderInfoDS = orderInfo.map(new MapFunction<String, Order_info>() {
            @Override
            public Order_info map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject data = jsonObject.getJSONObject("data");
                return JSON.parseObject(data.toJSONString(), Order_info.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Order_info>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                //create_time=2022-07-12T02:54:36Z
                .withTimestampAssigner((event, timestamp) -> {
//                    String s = event.getCreate_time().replace("T", " ");
//                    //System.out.println(s);
//                    String ss = s.replace("Z", "");
//                    //System.out.println(ss);
                    try {
                        return sdf.parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
        //orderInfoDS.print();

        KeyedStream<Order_info, String> keyDS = orderInfoDS.keyBy(x -> x.getOrder_no());

        //keyDS.print();
        Pattern<Order_info, Order_info> pattern = Pattern.<Order_info>begin("start").where(
                new SimpleCondition<Order_info>() {
                    @Override
                    public boolean filter(Order_info event) {
                        return event.getStatus().equals("60010");
                    }
                }
        ).next("middle").where(
                new SimpleCondition<Order_info>() {
                    @Override
                    public boolean filter(Order_info subEvent) {
                        return subEvent.getStatus().equals("60020");
                    }
                }
        ).within(Time.minutes(20));

        PatternStream<Order_info> pattern1 = CEP.pattern(keyDS, pattern);

        OutputTag<Order_info> side = new OutputTag<Order_info>("side") {};
        SingleOutputStreamOperator<Order_info> select = pattern1.select(side,
                new PatternTimeoutFunction<Order_info, Order_info>() {
                    @Override
                    public Order_info timeout(Map<String, List<Order_info>> pattern, long timeoutTimestamp) throws Exception {
                        List<Order_info> start = pattern.get("start");
                        return start.get(0);
                    }
                }, new PatternSelectFunction<Order_info, Order_info>() {
                    @Override
                    public Order_info select(Map<String, List<Order_info>> pattern) throws Exception {
//                        List<Order_info> start = pattern.get("start");
//                        return start.get(0);
                        return null;
                    }
                });
        select.print("支付成功");
        select.getSideOutput(side).print("超过20分钟支付 报警：-------");

        SingleOutputStreamOperator<String> process = orderInfoDS.windowAll(TumblingEventTimeWindows.of(Time.minutes(5)))
                .process(new ProcessAllWindowFunction<Order_info, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<Order_info, String, TimeWindow>.Context context, Iterable<Order_info> elements, Collector<String> out) throws Exception {
                        double price = 0.0;
                        int cnt = 0;
                        for (Order_info one : elements) {
                            price += Double.parseDouble(one.getAmount());
                            cnt++;
                        }
                        System.out.println(context.window().getStart() + "-" + context.window().getEnd() + "  5分钟之内所有订单的平均消费金额:" + (price / cnt));
                    }
                });
        process.print();


        env.execute();
    }
}
